package com.zhaopin.pulsar;

import com.zhaopin.common.container.RequestID;
import com.zhaopin.pulsar.extend.ProducerCreator;
import org.apache.commons.io.FileUtils;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.PulsarClientImpl;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 需要3个参数
 * 发送单个文件
 *
 * @author Administrator
 * @create 2019-05-15 17:22
 */
public class PulsarClientFile {


    public static void main(String[] args) throws IOException {
        //暂时先不导入数据

//        String topic_insert = "persistent://platform/dp/rankingword.update";
        //topic
        String topic_insert = args[0];

        //pulsar serUrl
        String serviceUrl = args[1];

        // 需要读取的json 格式的数据
        String filePath = args[2];

        //读取文件的list列表
        List<String> list = FileUtils.readLines(new File(filePath), "utf-8");

        PulsarClient client = PulsarClient.builder()
                .maxNumberOfRejectedRequestPerConnection(50)
                .serviceUrl(serviceUrl)
                .statsInterval(1L, TimeUnit.SECONDS)
                .build();

        ProducerCreator<byte[]> creator = new ProducerCreator((PulsarClientImpl) client, Schema.BYTES);
        Producer<byte[]> producer = creator.topic(topic_insert)
                .sendTimeout(5000000, TimeUnit.MILLISECONDS)
                .tracing(true)
                .metricsCollect(true)
                .create();

        //发送数据
        int i= 0;
        for (String s : list) {
            MessageId messageId = producer.send(s.getBytes("utf-8"));
            i++;
            System.out.println(i +" "+ topic_insert + "," + messageId + ", " + RequestID.getRequestID());
        }
        System.exit(1);
    }
}
